perf: plan take operations through FilteredReadExec's range-read path#7672
perf: plan take operations through FilteredReadExec's range-read path#7672LuQQiu wants to merge 22 commits into
Conversation
TakeExec reads rows through the v2 readers' point-lookup path with serial
per-fragment opens; reading the same rows through FilteredReadExec's planned
range reads is 2.3-2.7x faster (take-100 benchmark, warm NVMe).
Generalize FilteredReadExec's input to "a plan that tells the node which rows
to read", accepting two encodings detected from the input plan's schema:
- the serialized IndexExprResult mask batch (existing behavior, unchanged)
- any plan carrying a _rowid/_rowaddr column ("take mode", new): each batch
is converted to an in-memory IndexExprResult and read through the same
plan_scan -> read_fragment pipeline, then the columns are merged back into
the input batch preserving row order, duplicates, and payload columns
(e.g. _distance) - the same contract as TakeExec
Scanner::take and merge_insert now plan takes as FilteredReadExec on the v2
storage format. TakeExec is unchanged and still used for legacy (v1) storage.
The CoalesceTake optimizer rule also collapses stacked take-mode reads, and
count-pushdown / distributed proto serialization explicitly skip take-mode
nodes (distributed support is a follow-up).
Adds a three-arm benchmark (TakeExec vs mask mode vs take mode) at
rust/lance/benches/take_exec_compare.rs.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
gen_range is deprecated and fails clippy -D warnings. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Box the TakeRows variant (large size difference between variants) and allow print_stdout in the benchmark like the other benches. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
A random take touches many fragments (often one row per fragment) and try_join_all drives the opens concurrently but on a single task, so the CPU-bound part of ~100 fragment opens ran back to back (~4ms/query on a many-fragment dataset). Spawn one task per fragment open and per decode, matching the scan path (FilteredReadStream::try_new). Local take-100 (2M rows, 20 fragments): take mode 671 -> 858 QPS, now within 7% of mask mode (920 QPS) vs TakeExec at 445 QPS. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
github-actions
Bot
added
A-python
Codecov Report❌ Patch coverage is 📢 Thoughts on this report? Let us know! |
Review feedback: the node read as if it had "modes" (mode=take display, is_take() checks). Reframe it as one general I/O node, output = read(row_source, fields_to_read) + carry_columns, where only the row source varies: AllRows (no input), RowSet (one serialized IndexExprResult batch; sets are unordered and unique by nature), or RowStream (record batches whose order, duplicates, and columns are preserved). The source kind is derived from the input plan's schema, never configured. Renames only, no behavior change: FilteredReadInput -> RowSource, TakeRowsInput -> RowStreamSource, is_take() -> row_stream_input(), mode=take(_rowid) -> source=stream(_rowid) in plan text, and consumer checks (count pushdown, proto, CoalesceTake) now ask attributes instead of node identity. Also documents the legacy-storage invariant at both TakeExec swap sites (the v1 reader cannot serve FilteredRead) and the deferred count-through-streams rewrite in count_pushdown.rs. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Review feedback: the three-arm comparison harness is a local tool, not a maintained benchmark. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
westonpace
left a comment
westonpace
left a comment
There was a problem hiding this comment.
Initial feedback. Some of these comments are pretty verbose and I'd prefer to trim them down if possible. Overall I think the strategy is fine.
Also, why can't we use row addresses as input unless we have stable row id? I would think we could use addresses or ids as input with or without stable row id.
| /// Who names the rows that a [`FilteredReadExec`] reads. | ||
| /// | ||
| /// The node itself is a single, general I/O operator: | ||
| /// | ||
| /// ```text | ||
| /// output = read(row_source, fields_to_read) ⊕ carry_columns | ||
| /// | ||
| /// schema() = carry_columns + fields_to_read (uniform for every source) | ||
| /// ``` | ||
| /// | ||
| /// Only the row source varies, and the differences between the variants are | ||
| /// inherent to what the sources *are* — not modes of the node: | ||
| /// | ||
| /// - [`AllRows`](Self::AllRows): no input plan; every live row. No carry | ||
| /// columns. | ||
| /// - [`RowSet`](Self::RowSet): a *set* of rows, delivered as exactly one | ||
| /// batch in the serialized [`IndexExprResult`] wire layout (produced by | ||
| /// e.g. [`ScalarIndexExec`](super::scalar_index::ScalarIndexExec) or a | ||
| /// `_rowid IN (...)` filter). Sets are unordered and unique by nature, so | ||
| /// the output is in storage order and deduplicated. No carry columns. | ||
| /// - [`RowStream`](Self::RowStream): a *stream* of rows — ordinary record | ||
| /// batches with a `_rowid`/`_rowaddr` column. Streams have row order, | ||
| /// duplicates, and their own columns, and all three are preserved: the | ||
| /// stream's columns are carried through to the output next to the newly | ||
| /// read fields (the contract [`TakeExec`](super::TakeExec) provides). | ||
| /// Internally each batch is converted to an in-memory [`IndexExprResult`], | ||
| /// so a stream is the streaming generalization of a set. | ||
| /// | ||
| /// The source kind is *derived*, never configured: [`FilteredReadExec::try_new`] | ||
| /// inspects the input plan's schema (the wire layouts are unmistakable — | ||
| /// binary mask columns; anything else must carry a row id or row address | ||
| /// column). |
There was a problem hiding this comment.
This comment seems excessively verbose and is hard to parse. I like the description of each of the options, but those descriptions should be on the enum choices themselves, and not in this header. Maybe just something like:
/// Describes which rows should be read
///
/// This can be all rows, a specific set of rows, or the read can
/// be a "take" which reads new columns into an existing set of rows
/// using the `_rowid` or `_rowaddr`.
| /// | ||
| /// `index_input` generalizes to "a plan that tells this node which rows to | ||
| /// read" and accepts two encodings, detected from the plan's schema: | ||
| /// | ||
| /// - A serialized [`IndexExprResult`] batch (the wire layout emitted by | ||
| /// [`ScalarIndexExec`](super::scalar_index::ScalarIndexExec)): a row | ||
| /// *set* scoping a scan. This is the classic behavior. | ||
| /// - Any other plan carrying a `_rowid` or `_rowaddr` column: a row | ||
| /// *stream* of rows. The node reads `options.projection` for exactly | ||
| /// those rows and merges the columns into the stream's batches, | ||
| /// preserving row order, duplicates, and the stream's own columns — | ||
| /// the same contract as [`TakeExec`](super::TakeExec). Fields already | ||
| /// present in the stream's schema are not re-read. |
There was a problem hiding this comment.
This comment feels pretty repetitive. Also, we should rename index_input since it isn't necessarily index-specific anymore. Perhaps just:
/// Create a new filtered read
///
/// `input` identifies which rows to read. This is parsed into
/// a [`RowStreamSource`]
| if dataset.manifest.uses_stable_row_ids() { | ||
| return Err(Error::invalid_input_source( | ||
| format!( | ||
| "cannot read rows by '{}' on a dataset with stable row ids; the input plan must provide '{}'", | ||
| ROW_ADDR, ROW_ID | ||
| ) | ||
| .into(), | ||
| )); | ||
| } | ||
| ROW_ADDR |
There was a problem hiding this comment.
Why not? It almost seems like this would be the easier case since row addresses are what we ultimately need to locate the rows.
| // 4 - Take the mapped row ids. On the v2 storage format the take is | ||
| // planned as a FilteredReadExec fed by the index-mapper stream: | ||
| // the row ids become a row-id mask read through the planned | ||
| // range-read path, which is considerably faster than TakeExec's | ||
| // point-lookup path. Both nodes have the same output contract | ||
| // (input columns first, then the fetched columns). | ||
| // | ||
| // INVARIANT: every site that replaces TakeExec with | ||
| // FilteredReadExec must check is_legacy_storage() first. The v1 | ||
| // reader's read_ranges_tasks is a stub that errors ("Attempt to | ||
| // perform FilteredRead on v1 files"), so TakeExec remains the | ||
| // only take that can read v1 files. |
There was a problem hiding this comment.
Do we really need this verbose comment?
There was a problem hiding this comment.
I will clean up the verbose comments after all the code changes looks good (otherwise agent tend to add back a lot of comments when i delete them up front lollll)
| // A read with a row-stream source emits one row per input row; its count | ||
| // is driven by the input plan, not by fragment metadata. | ||
| // | ||
| // COUNT over such a read == the count of input rows whose id still | ||
| // exists. A future CountLiveRowsExec could answer that without any | ||
| // column I/O: stream the input and, per batch, build the live-row set | ||
| // from fragment metadata + deletion vectors and count member rows | ||
| // (per row, so duplicates count). Deferred because no plan shape today | ||
| // places a row-stream read under a COUNT — count plans request no | ||
| // columns, so the scanner never builds one. |
There was a problem hiding this comment.
| // A read with a row-stream source emits one row per input row; its count | |
| // is driven by the input plan, not by fragment metadata. | |
| // | |
| // COUNT over such a read == the count of input rows whose id still | |
| // exists. A future CountLiveRowsExec could answer that without any | |
| // column I/O: stream the input and, per batch, build the live-row set | |
| // from fragment metadata + deletion vectors and count member rows | |
| // (per row, so duplicates count). Deferred because no plan shape today | |
| // places a row-stream read under a COUNT — count plans request no | |
| // columns, so the scanner never builds one. | |
| // We don't currently support count pushdown when the row selector | |
| // is a row stream. |
| // One read round per input batch: read each fragment's rows as a | ||
| // single batch, and prioritize earlier input batches so consuming | ||
| // the (ordered) output is not starved by later batches | ||
| scoped.priority = batch_index; |
There was a problem hiding this comment.
Hmm, plan_to_scoped_fragments is already going to assign a priority (based on fragment order). I'd rather not lose this. In other words, if we have 4 input batches and 2 fragments then instead of using priorities 0, 0, 1, 1, 2, 2, 3, 3 I'd rather see 0, 1, 2, 3, 4, 5, 6, 7.
Can batch_index be a counter instead so we can do something like...
for scoped in &mut scoped_fragments {
scoped.priority += batch_index;
}
batch_index += scoped_fragments.len();
| ) | ||
| }) | ||
| .boxed() | ||
| .try_buffered(get_num_compute_intensive_cpus()) |
There was a problem hiding this comment.
This probably should not be get_num_compute_intensive_cpus as we are not really trying to split up CPU work. It's not really I/O work either. It's more of a "prefetch" factor.
The case for a large value, is when we have lots of input rows from a very fast source and a very small number of fragments. For example, maybe we have 1M input rows, available almost immediately, in batches of 8Ki, and each batch hits one fragment. Without enough buffering here we might not get enough I/O parallelism.
The only case I see for a small value, is when we have carry-through columns that are large. In that case if we buffer too much, we will accumulate a lot of carry-through data, which creates a lot of RAM pressure.
Maybe for now we hard-code it to something like 64. I think, in all current cases, we are unlikely to get more than 1 or 2 input batches anyways, so it is a moot point.
If we do start to have large input streams in the future then we should tie it into the memory pool reservation system. We should try and grab a reservation for the carry-through data as each batch of input arrives. If we succeed, we spawn a map_batch call on it. If we fail, then we pause reading the input until we can get the reservation.
| // target stays at the scanner batch size because take output | ||
| // batches mirror the input batches, and `batch_size` is a | ||
| // documented maximum for output batch sizes. | ||
| let coalesced = Arc::new(CoalesceBatchesExec::new(input, self.get_batch_size())); |
There was a problem hiding this comment.
Should the coalescing be an implicit part of the filtered read? I guess it could work either way but I feel like coalescing is very important here.
| if let Some(fragments) = &self.fragments { | ||
| read_options = read_options.with_fragments(Arc::new(fragments.clone())); | ||
| } |
There was a problem hiding this comment.
What does a "take with fragments" mean? Do you have any tests that cover this case?
There was a problem hiding this comment.
Depends on how we do large take/scan?
we can do split then send, split the rows by fragment and send to worker, then no "take_with_fragments" need
or broadcast take (send all rows, and each worker depends on its own fragment ranges, do take with fragments)?
|
|
||
| // Align the read rows (dataset order, deduplicated) back to the | ||
| // input's row order via the key column | ||
| let read_keys = read_data |
There was a problem hiding this comment.
I believe the code in take had a fast path to skip alignment if it wasn't needed (the order was already correct). Is that happening here?
A row-stream read now gathers its input into rounds of batch_size rows (explicit option, then LANCE_DEFAULT_BATCH_SIZE, then a flat 8192) before planning each read: tiny batches merge so planning overhead amortizes, and oversized batches split so a round's decoded output stays bounded. The external CoalesceBatchesExec in Scanner::take is gone - the node sizes its own rounds, and every row-stream consumer (e.g. merge_insert) now gets coalescing for free. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
A physical address already encodes (fragment, offset), so an address-keyed round now builds its read ranges directly instead of translating through the row-id sequence (which would misread addresses as stable row ids - the reason this case was previously rejected). Row ids keep the mask path; addresses at deleted rows drop like stale keys. Scanner::take can now hand any keyed input to FilteredReadExec, shrinking the TakeExec fallback to legacy storage only. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Round reads now keep both levels of I/O-scheduler ordering (rounds strictly ordered, fragments in dataset order within a round) via a stride instead of flattening each round to one priority. The round pipeline depth is a named prefetch constant (64) rather than the CPU count - the work is I/O bound and the window is what keeps the scheduler saturated on long input streams. When a round's read returns exactly one row per input row with an identical key sequence (storage-ordered input), the hash-map alignment and permutation are skipped. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Renames per review: the enum is RowSelector, try_new's parameter is simply `input`, and the enum/constructor docs shrink to short forms with per-variant descriptions on the variants. The legacy-storage and count-pushdown comments trim to one-liners. Also adds a unit test for a fragment-scoped take (in-scope key read, out-of-scope key dropped). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
…ce state execute_round now reads as plan_round -> read_round -> attach_columns, each stage with its own doc. RowStreamSource keeps one projection (inside its read options) plus the derived new-fields schema instead of storing the field list twice. The RowSelector enum no longer appears in constructor signatures - each variant is built in exactly one place. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
…ctor Options apply uniformly across selectors: a row-stream read with with_deleted_rows resolves deleted keys and returns their still-stored data instead of dropping them like stale keys. Fragments load without their deletion vectors so the planned ranges cover deleted offsets, and the reader keeps the rows by nulling their row ids - which is also why the flag requires an address-keyed input: a null row id cannot be aligned, while an address stays valid until compaction. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
…eams with_row_id / with_row_addr now mean the same thing for every selector: the column is present in the output iff its flag is set. For a row stream that is - carried and requested: kept; missing and requested: synthesized by the read (row addresses from read position, row ids from the fragment's row id sequence - no I/O); carried and unrequested: stripped. Ordinary input columns always carry through. Scanner::take preserves whatever identity the input carries (downstream nodes may key off it; the final ProjectionExec trims for free) and passes synthesis requests through, so existing plans are unchanged. merge_insert's v2 path now asks the take to synthesize _rowaddr instead of planning a separate AddRowAddrExec. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughThe change adds row-stream support to ChangesRow-stream filtered reads
Estimated code review effort: 5 (Critical) | ~120 minutes Sequence Diagram(s)sequenceDiagram
participant Scanner
participant FilteredReadExec
participant Dataset
Scanner->>FilteredReadExec: submit row-stream take request
FilteredReadExec->>Dataset: read requested columns by row identity
Dataset-->>FilteredReadExec: return aligned batches
FilteredReadExec-->>Scanner: produce projected rows
Possibly related issues
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
🧹 Nitpick comments (2)
rust/lance/src/io/exec/filtered_read.rs (2)
2399-2413: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winAvoid
.unwrap()on task joins in library code.
open_task.await.unwrap()?anddecode_task.await.unwrap()?(andres.unwrap()inapply) panic on aJoinError(task panic/cancellation) rather than surfacing aDataFusionError. Consider mapping the join error into an error path (e.g..map_err(|e| DataFusionError::External(...))?or.expect("reason")at minimum) so a joined-task failure propagates instead of aborting the worker.As per coding guidelines: "Never use
.unwrap(),.expect(),panic!(), orassert!()in library code for fallible operations; use?withResult" and "Reserve.unwrap()for tests only."🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@rust/lance/src/io/exec/filtered_read.rs` around lines 2399 - 2413, The task join handling in the filtered read path is panicking on JoinError because `open_task.await.unwrap()?`, `decode_task.await.unwrap()?`, and the similar `res.unwrap()` in `apply` bypass error propagation. Update the relevant join points in `filtered_read.rs` to map join failures into a `DataFusionError` (or another surfaced error type) before using `?`, so failures in `SpawnedTask`/task joins are returned instead of aborting the worker. Use the existing symbols `open_task`, `decode_task`, `SpawnedTask::spawn`, and `apply` to locate and replace the unsafe unwraps.Source: Coding guidelines
2564-2585: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winUse
SpawnedTask::spawn(...).in_current_span()here too.tokio::task::spawndetaches when the stream is dropped, so buffered rounds can keep running after early cancellation, and the spawned work loses the current tracing span.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@rust/lance/src/io/exec/filtered_read.rs` around lines 2564 - 2585, The round execution in apply is spawning detached tasks with tokio::task::spawn, which can outlive stream cancellation and drop tracing context. Replace that spawn path with SpawnedTask::spawn(...).in_current_span() in the apply pipeline for execute_round so buffered rounds stay tied to the current span and stop cleanly on early cancellation.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Nitpick comments:
In `@rust/lance/src/io/exec/filtered_read.rs`:
- Around line 2399-2413: The task join handling in the filtered read path is
panicking on JoinError because `open_task.await.unwrap()?`,
`decode_task.await.unwrap()?`, and the similar `res.unwrap()` in `apply` bypass
error propagation. Update the relevant join points in `filtered_read.rs` to map
join failures into a `DataFusionError` (or another surfaced error type) before
using `?`, so failures in `SpawnedTask`/task joins are returned instead of
aborting the worker. Use the existing symbols `open_task`, `decode_task`,
`SpawnedTask::spawn`, and `apply` to locate and replace the unsafe unwraps.
- Around line 2564-2585: The round execution in apply is spawning detached tasks
with tokio::task::spawn, which can outlive stream cancellation and drop tracing
context. Replace that spawn path with SpawnedTask::spawn(...).in_current_span()
in the apply pipeline for execute_round so buffered rounds stay tied to the
current span and stop cleanly on early cancellation.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 13ba5078-0401-4ed9-93ab-6b6b0a630118
📒 Files selected for processing (10)
python/python/tests/test_mem_wal.pyrust/lance-namespace-impls/src/dir.rsrust/lance/src/dataset/mem_wal/memtable/flush.rsrust/lance/src/dataset/scanner.rsrust/lance/src/dataset/write/merge_insert.rsrust/lance/src/io/exec/count_pushdown.rsrust/lance/src/io/exec/filtered_read.rsrust/lance/src/io/exec/filtered_read_proto.rsrust/lance/src/io/exec/optimizer.rsrust/lance/src/io/exec/take.rs
…m path The fragment-metadata loading and I/O-scheduler construction that the row-stream path had copied from FilteredReadStream::try_new move into shared helpers (load_all_fragments, make_scan_scheduler); RowStreamRead keeps only its lazy OnceCell wrapper. RowStreamSource also stops storing the carried schema - it is derived from the input plan's schema where needed instead of being a fifth field. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
…treams The physical view now works with either key column: an id-keyed round reconstructs the reader-nulled ids from the read addresses through the in-memory row id sequences (which retain deleted ids until compaction), and the output _rowid carries the scan's tombstone marker - null exactly for deleted rows - whether the column was carried or synthesized. Addresses stay real. Truly nonexistent keys still drop, including rows of fully-deleted fragments, which leave the manifest entirely. Also covers the row-set selector: a scalar index that was not rebuilt after a delete still returns the deleted id - without the flag the deletion-aware planning drops it, with the flag the tombstone returns. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
The physical view through a take is not worth its implementation surface with no consumers: the reader reports deleted rows by nulling their row id, which forced an output-schema nullability rewrite, an id-vs-address asymmetry, read-side id reconstruction, and a post-merge tombstone marker. The flag is rejected at construction again; the scan and row-set selectors keep their existing with_deleted_rows behavior, including the stale-index coverage. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (5)
rust/lance/src/io/exec/filtered_read.rs (5)
1734-1749: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winValidate the row key type during construction.
A non-
UInt64_rowidor_rowaddrcurrently passestry_new_row_streamand later fails asDataFusionError::Internal. Reject it immediately withError::invalid_input_source, including the column name and actual type.Proposed validation
- let key_column = if input_schema.column_with_name(ROW_ID).is_some() { - ROW_ID - } else if input_schema.column_with_name(ROW_ADDR).is_some() { - ROW_ADDR + let (key_column, key_field) = + if let Some((_, field)) = input_schema.column_with_name(ROW_ID) { + (ROW_ID, field) + } else if let Some((_, field)) = input_schema.column_with_name(ROW_ADDR) { + (ROW_ADDR, field) } else { return Err(Error::invalid_input_source( format!( "a row-stream input plan must have a column named '{}' or '{}'", ROW_ADDR, ROW_ID ) .into(), )); }; + + if key_field.data_type() != &arrow_schema::DataType::UInt64 { + return Err(Error::invalid_input_source( + format!( + "row-stream key column '{}' must be UInt64, but was {}", + key_column, + key_field.data_type() + ) + .into(), + )); + }As per coding guidelines, caller data issues must use an invalid-input error and include relevant names and types.
Also applies to: 2261-2279
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@rust/lance/src/io/exec/filtered_read.rs` around lines 1734 - 1749, Validate the selected row key’s data type in try_new_row_stream immediately after choosing key_column; require DataType::UInt64 and otherwise return Error::invalid_input_source with the column name and actual type included in the message. Apply the same validation to the corresponding row-key selection logic around the alternate location, preserving the existing ROW_ID-over-ROW_ADDR preference.Source: Coding guidelines
1295-1323: 📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick winDocument the complete row-stream API contract with an example.
Add a
try_newexample showing_rowid/_rowaddrinput, projection, and output behavior. Also document thatonly_indexed_fragmentsis unsupported for row streams and link toFilteredReadOptionsandrow_stream_inputrather than only the privateRowSelector.As per coding guidelines, all public APIs must have documentation with examples and links to relevant structs and methods.
Also applies to: 1651-1658, 2088-2101
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@rust/lance/src/io/exec/filtered_read.rs` around lines 1295 - 1323, Complete the public row-stream API documentation around FilteredReadOptions::try_new and row_stream_input with a runnable example covering _rowid/_rowaddr input, projection, and resulting output behavior. Link to the public FilteredReadOptions and row_stream_input APIs rather than only the private RowSelector type. Explicitly document that only_indexed_fragments is unsupported for row-stream reads, and apply the same documentation updates to the related row-stream API sections.Source: Coding guidelines
2822-2838: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winReject empty child lists for row-stream / row-set plans
rust/lance/src/io/exec/filtered_read.rs:2822-2838
with_new_children(vec![])on aRowStreamrebuilds throughtry_new(..., None)and silently turns the node intoAllRows, dropping the input stream. Match onself.inputand require 0 children forAllRows, 1 forRowSet/RowStream.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@rust/lance/src/io/exec/filtered_read.rs` around lines 2822 - 2838, Update FilteredReadExec::with_new_children to validate child counts based on self.input: require zero children for AllRows, exactly one child for RowSet and RowStream, and reject all mismatches before rebuilding. For the valid one-child cases, pass the child to try_new; do not allow an empty list to call try_new with None and change a row-stream or row-set plan into AllRows.
2702-2708: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winDowngrade row-stream row-count precision.
RowStreamcan drop rows for null, stale, or unknown keys before merge, so copyingsource.plan.partition_statistics(partition)?.num_rowsthrough unchanged can overstate the output as exact. Return it as inexact (or absent) instead.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@rust/lance/src/io/exec/filtered_read.rs` around lines 2702 - 2708, Downgrade the RowStream row-count statistic in the RowSelector::RowStream branch: do not copy partition_statistics(partition)?.num_rows as an exact value because rows may be dropped before merging. Preserve the estimate only as inexact, or omit it, while keeping the remaining unknown statistics behavior unchanged.
2366-2379: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winMake the spawned read pipeline cancellation-safe. The
tokio::spawnround scheduler and theseunwrap()ed joins detach work when the stream is dropped, and any join failure will panic the pipeline. Drive the futures directly throughbuffered/try_bufferedand propagate task errors instead.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@rust/lance/src/io/exec/filtered_read.rs` around lines 2366 - 2379, The read pipeline is not cancellation-safe because SpawnedTask joins are unwrap-based and detached work can outlive the stream. In the code assembling read_tasks and decode_tasks, replace spawned task scheduling and direct join awaits with futures driven through buffered or try_buffered, and propagate join/task errors with ? rather than unwrap, ensuring dropping the stream cancels in-flight work.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@rust/lance/src/io/exec/filtered_read.rs`:
- Around line 5375-5386: Update the affected tests around
FilteredReadExec::try_new to destructure the returned error as
Error::InvalidInput { source, .. }, then assert the expected message on source
rather than only checking the formatted error string. Apply the same
variant-and-message assertions to the additional cases noted around lines
5403–5428.
---
Outside diff comments:
In `@rust/lance/src/io/exec/filtered_read.rs`:
- Around line 1734-1749: Validate the selected row key’s data type in
try_new_row_stream immediately after choosing key_column; require
DataType::UInt64 and otherwise return Error::invalid_input_source with the
column name and actual type included in the message. Apply the same validation
to the corresponding row-key selection logic around the alternate location,
preserving the existing ROW_ID-over-ROW_ADDR preference.
- Around line 1295-1323: Complete the public row-stream API documentation around
FilteredReadOptions::try_new and row_stream_input with a runnable example
covering _rowid/_rowaddr input, projection, and resulting output behavior. Link
to the public FilteredReadOptions and row_stream_input APIs rather than only the
private RowSelector type. Explicitly document that only_indexed_fragments is
unsupported for row-stream reads, and apply the same documentation updates to
the related row-stream API sections.
- Around line 2822-2838: Update FilteredReadExec::with_new_children to validate
child counts based on self.input: require zero children for AllRows, exactly one
child for RowSet and RowStream, and reject all mismatches before rebuilding. For
the valid one-child cases, pass the child to try_new; do not allow an empty list
to call try_new with None and change a row-stream or row-set plan into AllRows.
- Around line 2702-2708: Downgrade the RowStream row-count statistic in the
RowSelector::RowStream branch: do not copy
partition_statistics(partition)?.num_rows as an exact value because rows may be
dropped before merging. Preserve the estimate only as inexact, or omit it, while
keeping the remaining unknown statistics behavior unchanged.
- Around line 2366-2379: The read pipeline is not cancellation-safe because
SpawnedTask joins are unwrap-based and detached work can outlive the stream. In
the code assembling read_tasks and decode_tasks, replace spawned task scheduling
and direct join awaits with futures driven through buffered or try_buffered, and
propagate join/task errors with ? rather than unwrap, ensuring dropping the
stream cancels in-flight work.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 3500d99d-6277-47ba-b4f8-cda40a5eddc2
📒 Files selected for processing (1)
rust/lance/src/io/exec/filtered_read.rs
Rounds now spawn via SpawnedTask (cancelling the query aborts in-flight rounds, and the tracing span is kept); task-join failures surface as DataFusionError instead of panicking the worker; construction-error tests assert the error variant as well as the message. apply() also drops its six per-field clones for two named Arc clones and inlines the single-use round-target helper. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Every multi-line rationale block shrinks to the one or two lines that state a non-obvious constraint; narration and name-restating docs are removed. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
The take now prints as 'LanceRead ... source=stream' instead of TakeExec's '(column)' notation: late materialization is asserted via the row-stream projection, and 'no scan happened' via the absence of a scan-flavored read. Also unlinks two rustdoc references to the private RowSelector type. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (1)
python/python/tests/test_dataset.py (1)
4511-4520: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winParameterize the late-materialization plan cases.
The
None,False, andTruecases differ only by input and expected marker. Consolidating them with@pytest.mark.parametrizewill avoid duplicated scanner/assertion logic and keep future coverage aligned.As per coding guidelines, use
@pytest.mark.parametrizefor Python test cases that differ only by inputs.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@python/python/tests/test_dataset.py` around lines 4511 - 4520, The late-materialization assertions in the affected dataset test duplicate identical scanner and plan-check logic for None, False, and True. Add pytest.mark.parametrize with each value and its expected plan marker, then consolidate the cases into one test that invokes dataset.scanner(filter=filt, late_materialization=...) and asserts the corresponding marker.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@python/python/tests/test_dataset.py`:
- Around line 4514-4516: Update the assertion in the eager-projection test to
check for the exact plan fragment "projection=[filter, values]" rather than the
broad ", values" substring, ensuring it specifically verifies that values
remains in the projection.
In `@python/python/tests/test_scalar_index.py`:
- Around line 627-633: Add positive assertions for "source=stream" to the
explain-plan checks in both the vector-search and full-text-search cases, while
retaining the existing "num_fragments" absence checks and row-count assertion.
Update the relevant assertions around make_vec_search and make_fts_search to
verify the FilteredReadExec row-stream representation.
---
Nitpick comments:
In `@python/python/tests/test_dataset.py`:
- Around line 4511-4520: The late-materialization assertions in the affected
dataset test duplicate identical scanner and plan-check logic for None, False,
and True. Add pytest.mark.parametrize with each value and its expected plan
marker, then consolidate the cases into one test that invokes
dataset.scanner(filter=filt, late_materialization=...) and asserts the
corresponding marker.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: cbff6dc4-793e-4f24-b8d9-1f8be9ad1b66
📒 Files selected for processing (3)
python/python/tests/test_dataset.pypython/python/tests/test_scalar_index.pyrust/lance/src/io/exec/filtered_read.rs
🚧 Files skipped from review as they are similar to previous changes (1)
- rust/lance/src/io/exec/filtered_read.rs
| assert ", values" in dataset.scanner( | ||
| filter=filt, late_materialization=False | ||
| ).explain_plan(True) |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Make the eager-projection assertion specific.
", values" can match unrelated plan text. Assert the expected eager projection directly, such as "projection=[filter, values]", so the test proves values was not moved to the row-stream path.
Proposed fix
- assert ", values" in dataset.scanner(
+ assert "projection=[filter, values]" in dataset.scanner(
filter=filt, late_materialization=False
).explain_plan(True)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| assert ", values" in dataset.scanner( | |
| filter=filt, late_materialization=False | |
| ).explain_plan(True) | |
| assert "projection=[filter, values]" in dataset.scanner( | |
| filter=filt, late_materialization=False | |
| ).explain_plan(True) |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@python/python/tests/test_dataset.py` around lines 4514 - 4516, Update the
assertion in the eager-projection test to check for the exact plan fragment
"projection=[filter, values]" rather than the broad ", values" substring,
ensuring it specifically verifies that values remains in the projection.
| assert "num_fragments" not in plan # no scan; the take prints as LanceRead | ||
| assert make_vec_search(ds).to_table().num_rows == 6 | ||
|
|
||
| plan = make_fts_search(ds).explain_plan() | ||
| assert "ScalarIndexQuery" in plan | ||
| assert "KNNVectorDistance" not in plan | ||
| assert "LanceRead" not in plan | ||
| assert "num_fragments" not in plan # no scan; the take prints as LanceRead |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Assert the new row-stream representation positively.
Checking only that "num_fragments" is absent does not prove the take used the intended FilteredReadExec row-stream path. Add an assertion for "source=stream" (while retaining the existing negative check and row-count assertion) in both vector and full-text cases.
Proposed fix
assert "num_fragments" not in plan
+ assert "source=stream" in plan
assert make_vec_search(ds).to_table().num_rows == 6
@@
assert "num_fragments" not in plan
+ assert "source=stream" in plan
assert make_fts_search(ds).to_table().num_rows == 6📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| assert "num_fragments" not in plan # no scan; the take prints as LanceRead | |
| assert make_vec_search(ds).to_table().num_rows == 6 | |
| plan = make_fts_search(ds).explain_plan() | |
| assert "ScalarIndexQuery" in plan | |
| assert "KNNVectorDistance" not in plan | |
| assert "LanceRead" not in plan | |
| assert "num_fragments" not in plan # no scan; the take prints as LanceRead | |
| assert "num_fragments" not in plan # no scan; the take prints as LanceRead | |
| assert "source=stream" in plan | |
| assert make_vec_search(ds).to_table().num_rows == 6 | |
| plan = make_fts_search(ds).explain_plan() | |
| assert "ScalarIndexQuery" in plan | |
| assert "KNNVectorDistance" not in plan | |
| assert "num_fragments" not in plan # no scan; the take prints as LanceRead | |
| assert "source=stream" in plan |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@python/python/tests/test_scalar_index.py` around lines 627 - 633, Add
positive assertions for "source=stream" to the explain-plan checks in both the
vector-search and full-text-search cases, while retaining the existing
"num_fragments" absence checks and row-count assertion. Update the relevant
assertions around make_vec_search and make_fts_search to verify the
FilteredReadExec row-stream representation.
…xecutor FilteredReadStream construction is parameterized (new_shared) so a round injects its per-query scheduler, cached fragment metadata, and a priority offset, then drains the ordinary scan pipeline into one batch. The row-stream path's hand-rolled open/drain/decode loops are deleted; the fragment-readahead cap, decode buffering, spans, and error handling are the scan's own code. RowStreamSource is shared via Arc, so RowStreamRead holds it directly instead of flattened copies of its fields. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
The row-stream unit of work is called a batch throughout (execute_batch, plan_batch, read_batch, coalesce_batches, ROW_STREAM_PREFETCH_BATCHES); docs revert to their pre-PR wording where the original said it better, and the remaining review-flagged comments are removed or reworded. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
The collapse rebuilt the outer take against the inner take's input, but FilteredReadExec stored a pre-subtracted projection, so the rebuild lost the inner take's columns and the name-based remap panicked. Fix both directions: skip the transform whenever the rebuilt schema would drop a column of the original output, and have Scanner::take pass the full un-subtracted output projection (the node subtracts internally) so the rebuild re-derives what to fetch, like TakeExec. Also fail the reorder check when field counts differ instead of zipping past the shorter list. Adds dedicated CoalesceTake tests (row-stream pair, guard case, legacy). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2 participants
Motivation
TakeExecmaterializes rows through the v2 readers' point-lookup path (ReadBatchParams::Indices)with per-fragment reader opens driven on a single task. Reading the same rows through
FilteredReadExec's planned range reads is much faster. Take-100 benchmark (100 random rowaddresses per query, warm NVMe, heavy ~1KiB string column; harness kept out of the tree):
_rowid IN (...))Carrying a payload column (e.g.
_distance/_score) through the row-stream read costs ~2% —the capability a row-set input cannot express.
Design: one I/O node, three row selectors
FilteredReadExecstays a single, general I/O node. Which rows it reads is described by aRowSelector, derived from the input plan's schema at construction — never configured:IndexExprResultbatch (scalar index result,_rowid IN) — existing behavior, unchanged_rowid/_rowaddrcolumn_distance), preservedA row stream executes in rounds: the input is coalesced to
batch_sizerows per round(explicit option →
LANCE_DEFAULT_BATCH_SIZE→ 8192; merging small batches amortizesplanning, slicing big ones bounds a round's memory), each round is planned and read through
the same
plan_scan→plan_to_scoped_fragments→read_fragmentpipeline as a set, andthe fetched columns are aligned back to the round's row order (hash on the key column, O(N);
skipped entirely when the read comes back already aligned) and merged in — including
nested-struct merges, via the same
calculate_output_schema+merge_with_schemacontractas
TakeExec. Up to 64 rounds run concurrently (a prefetch depth, cancelled with the query);I/O priorities keep rounds strictly ordered and fragments in dataset order within a round.
Semantics of the row-stream selector:
_rowid(preferred) or_rowaddr, on both stable- and non-stable-row-iddatasets. Ids resolve through the fragments' row-id sequences; addresses resolve directly
by position.
_rowid/_rowaddrappear in the output iff theirprojection flag is set — carried columns kept, missing ones synthesized by the read (no
I/O: addresses come from read position, ids from the in-memory sequences), unrequested
carried ones stripped. This lets merge_insert drop its
AddRowAddrExecnode.with_deleted_rows(plus filters, scanranges,
only_indexed_fragments) is rejected at construction — the reader marks deletedrows by nulling their row id, which cannot be aligned back to input keys.
Call sites:
Scanner::take()and merge_insert's indexed take plan takes asFilteredReadExecon the v2 storage format; the scanner's externalCoalesceBatchesExecisgone (the node coalesces its own input). Plan text shows
LanceRead: ..., projection=[...], source=stream(_rowid)whereTake: columns=...appearedbefore.
Design notes (from review)
read_ranges_tasksis a stub that errors —FilteredReadExecphysically cannot read v1files. v1 datasets keep bit-for-bit yesterday's behavior; TakeExec becomes deletable when
v1 read support is dropped.
input plan, not fragment metadata. Implementable without column I/O (per-batch liveness
count against fragment metadata + deletion vectors), deferred because no plan shape places
a row-stream read under a COUNT today.
(TakeExec itself sorted and inverse-permuted); duplicates and stable-row-id address order
make request-order reads impossible. Alignment is O(N) per round, ~2% of query time, and
skipped when the round is already in storage order.
Behavior change
Input rows whose row id/address no longer exists (stale index results pointing at deleted
rows) are now silently dropped on non-stable-row-id datasets, where
TakeExecfailed with arow-count mismatch error. This matches the existing stable-row-id behavior and FTS
deleted-row expectations.
Follow-ups (out of scope, marked in code)
filtered_read_exec_to_protoreturnsnot_supported).NULL
_rowidmemtable rows).memory-pool-based round admission.
constant-memory diagonal with a random-scatter key workload.
Testing
aligned fast path, fragment scoping, identity-flag matrix (keep/synthesize/strip, both
directions), nested struct merge, stable row ids, stale-id drops,
with_deleted_rowsrejection, construction errors,
with_new_children, execution without a tokio runtime.with_deleted_rowscoverage (a deleted id still in an un-rebuiltindex drops on the live view and returns as a null-
_rowidtombstone on the physicalview).
lancelib suite,--features slow_testsquery integration suite,lance-namespace-impls; plan-text assertions updated (Rust + Python).🤖 Generated with Claude Code
Summary by CodeRabbit
Improvements
Bug Fixes